package com.raymond.http.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.pool.ChannelPool;
import io.netty.channel.pool.ChannelPoolHandler;
import io.netty.channel.pool.SimpleChannelPool;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.ThrowableUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.util.Deque;
import java.util.Set;
import java.util.concurrent.*;

import static io.netty.util.internal.ObjectUtil.checkNotNull;

/**
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-12-15 18:53
 */
@Slf4j
public class HttpChannelPool implements Closeable {


    private final ConcurrentHashMap<ChannelId, Boolean> channelMap = new ConcurrentHashMap<>();

    private final Deque<Channel> deque = PlatformDependent.newConcurrentDeque();

    private final Bootstrap bootstrap;
    private final ChannelPoolHandler handler;
    public final Semaphore semaphore;

    private final boolean lastRecentUsed;

    public HttpChannelPool(Bootstrap bootstrap, ChannelPoolHandler handler, int maxConnection, boolean lastRecentUsed) {
        this.handler = checkNotNull(handler, "handler");
        this.bootstrap = checkNotNull(bootstrap, "bootstrap").clone();
        this.bootstrap.handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                assert ch.eventLoop().inEventLoop();
                handler.channelCreated(ch);
            }
        });
        this.lastRecentUsed = lastRecentUsed;
        this.semaphore = new Semaphore(maxConnection);
    }


    protected Bootstrap bootstrap() {
        return bootstrap;
    }


    protected ChannelPoolHandler handler() {
        return handler;
    }


    protected Channel pollChannel() {
        return lastRecentUsed ? deque.pollLast() : deque.pollFirst();
    }


    protected boolean offerChannel(Channel channel) {
        return deque.offer(channel);
    }

    protected boolean removeChannel(Channel channel) {
        return deque.remove(channel);
    }


    protected ChannelFuture connectChannel(Bootstrap bs) {
        return bs.connect();
    }


    public Channel acquire() {
        return acquire(-1L, TimeUnit.SECONDS);
    }

    public Channel acquire(long timeOut, TimeUnit timeUnit) {
        return acquire(bootstrap.config().group().next().newPromise(), timeOut, timeUnit);
    }

    private boolean acquire0(long timeOut, TimeUnit timeUnit) throws InterruptedException {
        if (timeOut <= 0) {
            semaphore.acquire();
            return true;
        } else {
            return semaphore.tryAcquire(timeOut, timeUnit);
        }

    }


    public Channel acquire(Promise<Channel> promise, long timeOut, TimeUnit timeUnit) {
        try {
            if (!acquire0(timeOut, timeUnit)) {
                throw new RuntimeException("获取连接超时");
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        Channel ch = pollChannel();
        if (ch == null) {
            try {
                ch = getChannel(promise, timeOut, timeUnit);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                semaphore.release();
                throw new RuntimeException("获取连接失败:", e);
            }
        }
        channelMap.put(ch.id(), true);
        return ch;
    }

    private Channel getChannel(Promise<Channel> promise, long timeOut, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
        Bootstrap bs = bootstrap.clone();
        ChannelFuture f = connectChannel(bs);
        if (f.isDone()) {
            notifyConnect(f, promise);
        } else {
            f.addListener((ChannelFutureListener) future -> notifyConnect(future, promise));
        }
        Channel channel;
        if (timeOut <= 0) {
            channel = promise.get();
        } else {
            channel = promise.get(timeOut, timeUnit);
        }
        return channel;
    }

    private void notifyConnect(ChannelFuture future, Promise<Channel> promise) {
        if (future.isSuccess()) {
            Channel channel = future.channel();
            if (!promise.trySuccess(channel)) {
                release(channel);
            }
        } else {
            promise.tryFailure(future.cause());
        }
    }


    public void release(Channel channel) {
        try {
            releaseAndOffer(channel);
        } catch (Exception e) {
            channel.close();
        }
    }

    public void close(Channel channel) {
        Boolean flag = channelMap.put(channel.id(), false);
        if (flag != null && flag) {
            semaphore.release();
        }
        removeChannel(channel);
        channel.close();
    }

    private void releaseAndOffer(Channel channel) throws Exception {
        Boolean flag = channelMap.put(channel.id(), false);
        if (flag != null && !flag) {
            return;
        }
        try {
            if (offerChannel(channel)) {
                handler.channelReleased(channel);
            } else {
                channel.close();
            }
        } finally {
            semaphore.release();
        }
    }

    public int availablePermits() {
        return semaphore.availablePermits();
    }

    @Override
    public void close() {
        for (;;) {
            Channel channel = pollChannel();
            if (channel == null) {
                break;
            }
            channel.close().awaitUninterruptibly();
        }
    }


}
